package com.fwmagic.flink.projectcase.queryactivitycase.map;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.JSONPath;
import com.fwmagic.flink.projectcase.queryactivitycase.bean.ActivityBean;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.http.HttpStatus;
import org.apache.http.client.HttpClient;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;
import org.apache.http.util.EntityUtils;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;

public class GPSToActivityBeanMapFunction extends RichMapFunction<String, ActivityBean> {

    private CloseableHttpClient httpClient;

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        httpClient = HttpClients.createDefault();
    }

    @Override
    public ActivityBean map(String line) throws Exception {
        //u001,A1,2019-09-02 10:10:11,1,115.908923,39.267291
        String[] fields = line.split(",");
        String uid = fields[0];
        String aid = fields[1];
        String time = fields[2];
        int eventType = Integer.parseInt(fields[3]);
        double longitude = Double.parseDouble(fields[4]);
        double latitude = Double.parseDouble(fields[5]);

        String province = null;

        String url = "https://restapi.amap.com/v3/geocode/regeo?output=json&location=" + longitude + "," + latitude + "&key=0cab9004c4e4c4c81966b2d85564cf96";
        //调用高德地图gps的restapi查询接口
        HttpGet httpGet = new HttpGet(url);
        CloseableHttpResponse response = null;

        response = httpClient.execute(httpGet);
        try {
            if (response != null && response.getStatusLine() != null
                    && response.getStatusLine().getStatusCode() == HttpStatus.SC_OK) {
                //获取请求的json串
                String result = EntityUtils.toString(response.getEntity());
                //获取城市
                //String city = JSONPath.read(result, "$.regeocode.addressComponent.city").toString();
                System.out.println("result:"+result);
                //获取省份
                province = JSONPath.read(result, "$.regeocode.addressComponent.province").toString();
            }
        } finally {
            response.close();
        }

        return ActivityBean.of(uid, aid, "", time, eventType, province,longitude,latitude);
    }

    @Override
    public void close() throws Exception {
        super.close();
    }
}
